package com.hmall.common.utils;

import cn.hutool.core.lang.UUID;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.util.concurrent.ListenableFutureCallback;


@Slf4j
@RequiredArgsConstructor
public class RabbitMQHelper {
    private final RabbitTemplate rabbitTemplate;

    /**
     * 发送消息，并设置用户信息
     * @param exchange
     * @param routingKey
     * @param message
     */
    public void sendWithUser(String exchange, String routingKey, Object message) {
        Long userId = UserContext.getUser();
        rabbitTemplate.convertAndSend(exchange, routingKey, message, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setHeader("user-info", userId);
                return message;
            }
        });
    }


    /**
     * 发送消息
     * @param exchange
     * @param routingKey
     * @param msg
     */
    public void sendMessage(String exchange, String routingKey, Object msg){
        log.debug("准备发送消息, exchange:{}, routionKey:{}, msg:{}", exchange, routingKey, msg);
        rabbitTemplate.convertAndSend(exchange, routingKey, msg);
    }

    /**
     * 发送延迟消息
     * @param exchange
     * @param routingKey
     * @param msg
     * @param delay
     */
    public void sendDelayMessage(String exchange, String routingKey, Object msg, int delay){
        log.debug("准备发送延迟消息, exchange:{}, routionKey:{}, msg:{}, delay:{}", exchange, routingKey, msg, delay);
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                message.getMessageProperties().setDelay(delay);
                return message;
            }
        });
    }

    /**
     * 发送消息，并设置生产者confirm back(确认回调)
     * @param exchange
     * @param routingKey
     * @param msg
     * @param maxRetries
     */
    public void sendMessageWithConfirm(String exchange, String routingKey, Object msg, int maxRetries){
        log.debug("准备发送配置有生产者confirm back的消息, exchange:{}, routionKey:{}, msg:{}, maxRetries:{}", exchange, routingKey, msg, maxRetries);
        CorrelationData cd = new CorrelationData(UUID.randomUUID().toString(true));
        cd.getFuture().addCallback(new ListenableFutureCallback<CorrelationData.Confirm>() {
            int retryCount;
            @Override
            public void onFailure(Throwable ex) {
                log.error("mq处理future内部出错", ex);
            }

            @Override
            public void onSuccess(CorrelationData.Confirm result) {
                if(result != null && !result.isAck()) {
                    log.debug("收到nack，消息发送失败，已重试次数: {}", retryCount);
                    if(retryCount >= maxRetries) {
                        log.error("消息发送重试次数耗尽，发送失败");
                        return;
                    }
                    CorrelationData cd = new CorrelationData(UUID.randomUUID().toString(true));  // toString(true)看看源码就懂了
                    cd.getFuture().addCallback(this);
                    rabbitTemplate.convertAndSend(exchange, routingKey, msg, cd);
                    retryCount++;
                }
            }
        });
        rabbitTemplate.convertAndSend(exchange, routingKey, msg, cd);
    }
}

